package drds.server.handler;

import java.util.concurrent.atomic.AtomicBoolean;

import drds.server.api_warpper.Lock;
import drds.server.config.ErrorCode;
import drds.server.datasource.IConnectionHolder;
import drds.server.net.packet.ErrorPacket;
import drds.server.server.Session;
import drds.server.util.StringUtil;

/**
 * 
 */
abstract class MultiNodeHandler extends HandlerAdapter implements Terminatable {
	protected final Lock lock = new Lock();
	protected final Session session;
	private AtomicBoolean isFailed = new AtomicBoolean(false);
	protected volatile String errorMessage;
	protected byte packetId;
	protected final AtomicBoolean errorRepsponsed = new AtomicBoolean(false);

	public MultiNodeHandler(Session session) {
		if (session == null) {
			throw new IllegalArgumentException("session is null!");
		}
		this.session = session;
	}

	public void setFail(String errorMessage) {
		isFailed.set(true);
		this.errorMessage = errorMessage;
	}

	public boolean isFail() {
		return isFailed.get();
	}

	private int nodeCount;

	protected boolean canClose(IConnectionHolder connectionHolder, boolean tryErrorFinish) {

		// realse this connection if safe
		session.releaseConnectionIfSafe(connectionHolder, false);
		boolean allFinished = false;
		if (tryErrorFinish) {
			allFinished = this.decrementCountBy(1);
			this.tryErrorFinished(allFinished);
		}

		return allFinished;
	}

	protected void decrementCountToZero() {
		lock.lock();
		try {
			nodeCount = 0;
		} finally {
			lock.unlock();
		}
	}

	public void connectionError(IConnectionHolder ConnectionHolder, Throwable e) {
		final boolean canClose = decrementCountBy(1);
		if (canClose) {
			setFail("backend connect: " + e);
		}
		this.tryErrorFinished(canClose);
	}

	public void errorResponse(IConnectionHolder ConnectionHolder, byte[] errorBytes) {
		session.releaseConnectionIfSafe(ConnectionHolder, false);
		ErrorPacket errorPacket = new ErrorPacket();
		errorPacket.read(errorBytes);
		String errorMessage = new String(errorPacket.errorMessage);
		this.setFail(errorMessage);

		this.tryErrorFinished(this.decrementCountBy(1));
	}

	public boolean clearIfSessionClosed(Session session) {
		if (session.closed()) {
			session.clearResources(true);
			return true;
		} else {
			return false;
		}

	}

	@SuppressWarnings("unused")
	protected boolean decrementCountBy(int finished) {
		boolean zeroReached = false;
		Runnable callback = null;
		lock.lock();
		try {
			if (zeroReached = --nodeCount == 0) {
			}
		} finally {
			lock.unlock();
		}
		if (zeroReached && callback != null) {
		}
		return zeroReached;
	}

	protected void reset(int initCount) {
		nodeCount = initCount;
		isFailed.set(false);
		errorMessage = null;
		packetId = 0;
	}

	protected ErrorPacket createErrorPacket(String errorMessage) {
		ErrorPacket errorPacket = new ErrorPacket();
		lock.lock();
		try {
			errorPacket.packetId = ++packetId;
		} finally {
			lock.unlock();
		}
		errorPacket.errorNo = ErrorCode.ER_UNKNOWN_ERROR;
		errorPacket.errorMessage = StringUtil.getBytes(errorMessage);
		return errorPacket;
	}

	protected void tryErrorFinished(boolean allEnd) {
		if (allEnd && !session.closed()) {
			if (errorRepsponsed.compareAndSet(false, true)) {
				createErrorPacket(this.errorMessage).write(session.getSessionContext());
			}
			// clear session resources,release all
			if (session.getSessionContext().isAutoCommit()) {
				session.closeAndClearResources(errorMessage);
			} else {
				session.getSessionContext().setTransactionInterruptStatus(this.errorMessage);
				// clear resouces
			}

		}

	}

	public void connectionClose(IConnectionHolder ConnectionHolder, String reason) {
		this.setFail("closed connection:" + reason + " con:" + ConnectionHolder);
		boolean finished = false;
		lock.lock();
		try {
			finished = (this.nodeCount == 0);

		} finally {
			lock.unlock();
		}
		if (finished == false) {
			finished = this.decrementCountBy(1);
		}
		if (errorMessage == null) {
			errorMessage = "back connection closed ";
		}
		tryErrorFinished(finished);
	}

}